package cn.itcast.map;

import cn.itcast.bean.IndexBean;
import cn.itcast.bean.IndexKlineBean;
import cn.itcast.bean.StockBean;
import cn.itcast.bean.StockKlineBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.functions.RichMapFunction;

import java.math.BigDecimal;
import java.math.RoundingMode;

/**
 * 个股K线数据转换,转换成Row
 * 通用数据转换方法(日K,周K,月K)
 * 本项目每天启停一次 (9:00, 16:00)
 */
public class IndexDayKlineMap extends RichMapFunction<IndexBean, IndexKlineBean> {



    /**
     * 二、业务处理
     * 1.获取个股部分数据（前收、收、开盘、高、低、量、金额）
     * 2.获取T日和周首次交易日时间,转换成long型
     * 3.比较周期首个交易日和当天交易日大小，判断是否是周、月K线
     * 4.获取周/月K数据：成交量、成交额、高、低
     * 5.高、低价格比较
     * 6.计算成交量、成交额
     * 7.计算均价
     * 8.封装数据Row
     */
    @Override
    public IndexKlineBean map(IndexBean value) throws Exception {
        //1.获取个股部分数据（前收、收、开盘、高、低、量、金额）
        //最新日K数据
        BigDecimal preClosePrice = value.getPreClosePrice();
        BigDecimal closePrice = value.getClosePrice();
        BigDecimal openPrice = value.getOpenPrice();
        BigDecimal highPrice = value.getHighPrice();
        BigDecimal lowPrice = value.getLowPrice();
        Long tradeVolDay = value.getTradeVolDay();
        Long tradeAmtDay = value.getTradeAmtDay();

        //2.获取T日和周期首次交易日时间,转换成long型
        String tradeDate = DateUtil.longTimeToString(value.getEventTime(), Constant.format_yyyy_mm_dd);

        //7.计算均价
        //tradeAmtDay / tradeVolDay
        BigDecimal avgPrice = new BigDecimal(0);
        if (tradeVolDay != 0) {
            avgPrice = new BigDecimal(tradeAmtDay).divide(new BigDecimal(tradeVolDay), 2, RoundingMode.HALF_UP);
        }

        //8.封装数据Row
        IndexKlineBean indexKlineBean = new IndexKlineBean(
                tradeDate,
                value.getIndexCode(),
                value.getIndexName(),
                preClosePrice,
                openPrice,
                highPrice,
                lowPrice,
                closePrice,
                tradeVolDay,
                tradeAmtDay,
                value.getEventTime(),
                value.getSource(),
                1
        );

        return indexKlineBean;
    }
}
